package com.atguigu.gmall.realtime.bootstrap

import java.util

import com.alibaba.fastjson.{JSONArray, JSONObject}
import com.atguigu.gmall.realtime.util.{MyKafkaSink, MySqlUtil}

object BootstrapDimApp {

  //目的 ： 把mysql的历史数据写入kafka
  // 直接执行 不需要spark环境
  def main(args: Array[String]): Unit = {

    val  dimTables= Array("user_info","base_province","sku_info")
    //1 、 读取mysql数据库
    for (tableName <- dimTables) {
      val sql=s"select * from $tableName"
      val jsonObjList: util.List[JSONObject] = MySqlUtil.queryList(sql)
      import  scala.collection.JavaConverters._
      //2 、调整结构  仿照canal的结构 转换格式
      for (dataJsonObj <- jsonObjList.asScala ) {
        val canalJsonObj = new JSONObject()
        val dataJsonArr = new JSONArray()
        dataJsonArr.add(dataJsonObj)
        canalJsonObj.put("data",dataJsonArr)
        canalJsonObj.put("database","gmall0701")
        canalJsonObj.put("table",tableName)
        canalJsonObj.put("type","INSERT")
        val dataKey=dataJsonObj.getString("id")
        //3 、写入kafka
        MyKafkaSink.send("ODS_BASE_DB_C",dataKey,canalJsonObj.toJSONString)
      }
      MyKafkaSink.flush()

    }

  }

}

